package com.mjf.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mjf.bean.VisitorStats;
import com.mjf.utils.ClickHouseUtil;
import com.mjf.utils.DateTimeUtil;
import com.mjf.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Date;
import java.util.concurrent.TimeUnit;

/**
 * 访客主题宽表计算
 *
 * 数据流向：web/app -> nginx -> springboot -> kafka(ods) -> flinkApp -> kafka(dwd) -> flinkApp -> kafka(dwm) -> flinkApp -> clickhouse(dws)
 * 程序：gmall2020-mock-log-2020-12-18.jar -> flink-logger.sh(包含 nginx/springboot/kafka(ods)) -> BaseLogApp(包含 flinkApp/kafka(dwd))
 *      -> UniqueVisitApp/UserJumpDetailApp(包含 flinkApp/kafka(dwm)) -> VisitorStatsApp(包含 flinkApp/clickhouse)
 */
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

/*
        // 测试时关闭
        // 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9000/gmall-flink/checkpoint"));
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);

        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10));
*/

        // 2.读取 Kafka 数据创建流
        String groupId = "VisitorStatsApp";
        String pageViewTopic = "dwd_page_log";
        String uniqueVisitTopic = "dwm_unique_visit";
        String userJumpDetailTopic = "dwm_user_jump_detail";
        String sinkTopic = "";

        // UV
        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(uniqueVisitTopic, groupId));
        // 用户跳出
        DataStreamSource<String> ujDS = env.addSource(MyKafkaUtil.getKafkaConsumer(userJumpDetailTopic, groupId));
        // PV
        DataStreamSource<String> pvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageViewTopic, groupId));

        // 3.将每个流处理成相同的数据类型
        // 3.1处理uv数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUvDS = uvDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            // 提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L, 0L, 0L, 0L, 0L,
                    jsonObject.getLong("ts")
            );
        });

        // 3.2处理uj数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithUjDS = ujDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            // 提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 0L, 0L, 1L, 0L,
                    jsonObject.getLong("ts")
            );
        });

        // 3.3处理pv数据
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithPvDS = pvDS.map(line -> {
            JSONObject jsonObject = JSON.parseObject(line);
            // 提取common字段
            JSONObject common = jsonObject.getJSONObject("common");
            // 获取页面信息
            JSONObject page = jsonObject.getJSONObject("page");

            String last_page_id = page.getString("last_page_id");
            long sv = 1L;   // 上一条页面ID为空时表示进入页面
            if (last_page_id != null && last_page_id.length() != 0) {
                sv = 0L;
            }

            return new VisitorStats(
                    "", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    0L, 1L, sv, 0L,
                    page.getLong("during_time"),
                    jsonObject.getLong("ts")
            );
        });

        // 4.union几个流
        DataStream<VisitorStats> unionDS = visitorStatsWithUvDS.union(visitorStatsWithUjDS, visitorStatsWithPvDS);

        // 5.提取时间戳生成watermark
        SingleOutputStreamOperator<VisitorStats> visitorStatsWithWM = unionDS.assignTimestampsAndWatermarks(
               // uj数据有延迟
                WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(11))
                        .withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
                            @Override
                            public long extractTimestamp(VisitorStats element, long recordTimestamp) {
                                return element.getTs();
                            }
                        })
        );

        // 6.按照维度信息分组
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyedStream = visitorStatsWithWM.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(VisitorStats value) throws Exception {
                return new Tuple4<String, String, String, String>(
                        value.getAr(),
                        value.getCh(),
                        value.getIs_new(),
                        value.getVc()
                );
            }
        });

        // 7.开窗聚合 10s的滚动聚合
        SingleOutputStreamOperator<VisitorStats> visitorStatsDS = keyedStream.window(TumblingEventTimeWindows.of(Time.of(10L, TimeUnit.SECONDS)))
                // reduce(reduceFunction, WindowFunction)   先增量聚合，再将聚合完的结果传给windowFunction
                .reduce(new ReduceFunction<VisitorStats>() {
                    @Override
                    public VisitorStats reduce(VisitorStats value1, VisitorStats value2) throws Exception {
                        return new VisitorStats(
                                value1.getStt(),
                                value1.getEdt(),
                                value1.getVc(),
                                value1.getCh(),
                                value1.getAr(),
                                value1.getIs_new(),
                                value1.getUv_ct() + value2.getUv_ct(),
                                value1.getPv_ct() + value2.getPv_ct(),
                                value1.getSv_ct() + value2.getSv_ct(),
                                value1.getUj_ct() + value2.getUj_ct(),
                                value1.getDur_sum() + value2.getDur_sum(),
                                value1.getTs()
                        );
                    }
                }, new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4,
                                      TimeWindow window, Iterable<VisitorStats> input, Collector<VisitorStats> out) throws Exception {
                        long start = window.getStart();
                        long end = window.getEnd();

                        VisitorStats visitorStats = input.iterator().next();
                        // 补充窗口信息
                        visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                        visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(end)));

                        out.collect(visitorStats);
                    }
                });

        // 8.将数据写入 clickhouse
        visitorStatsDS.print("visitorStatsDS=====>");
        visitorStatsDS.addSink(ClickHouseUtil.getSink(
                "insert into visitor_stats values(?,?,?,?,?,?,?,?,?,?,?,?)"
        ));

        // 9.启动任务
        env.execute(VisitorStatsApp.class.getName());

    }
}
